package com.example.sparkdemo.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 *
 * @author zhang.t.c
 * @date 2021/8/19
 */
object CustomReceiverStreamingTest {

  def main(args: Array[String]): Unit = {
    val sc: SparkConf = new SparkConf().setMaster("local[*]").setAppName("FileStream")
    val ssc = new StreamingContext(sc, Seconds(5))

    val lineStream: ReceiverInputDStream[String] = ssc.receiverStream(new CustomerReceiver("hadoop112", 9999))
    val wordcount = lineStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ +_)

    wordcount.print()

    ssc.start()
    ssc.awaitTermination()

  }

}
